package com.dong.common.mq;

import com.aliyun.openservices.ons.api.MessageListener;
import com.aliyun.openservices.ons.api.PropertyKeyConst;
import com.aliyun.openservices.ons.api.PropertyValueConst;
import com.aliyun.openservices.ons.api.bean.ConsumerBean;
import com.aliyun.openservices.ons.api.bean.ProducerBean;
import com.aliyun.openservices.ons.api.bean.Subscription;

import java.util.HashMap;
import java.util.Map;
import java.util.Properties;

/**
 * @author zhaodongchao
 * @date 2020/11/19 14:48
 */
public class MQServiceImpl implements MqService {
    @Override
    public ProducerBean createProducer(BaseConfig config) {
        Properties producerProperties = getPropertiesFromConfig(config, ConfigType.PRODUCER);
        ProducerBean producerBean = new ProducerBean();
        producerBean.setProperties(producerProperties);
        return producerBean;
    }

    @Override
    public ConsumerBean createConsumer(BaseConfig config, MessageListener messageListener) {
        if (null == messageListener) {
            throw new RuntimeException("消息监听器不能为空");
        }
        //获取配置文件
        Properties consumerProperties = getPropertiesFromConfig(config, ConfigType.CONSUMER);
        //新建ConsumerBean实例
        ConsumerBean consumer = new ConsumerBean();
        //设置配置文件
        consumer.setProperties(consumerProperties);

        //设置订阅关系
        Map<Subscription, MessageListener> subscriptionTable = new HashMap<>(2);
        Subscription subscription = new Subscription();
        subscription.setTopic(config.getTopic());
        subscription.setExpression(config.getExpression());
        subscriptionTable.put(subscription, messageListener);
        consumer.setSubscriptionTable(subscriptionTable);

        return consumer;
    }

    private Properties getPropertiesFromConfig(BaseConfig config, ConfigType configType) {
        //参数非空校验
        if (null == config) {
            throw new RuntimeException("配置不能为空");
        }
        Properties properties = new Properties();
        // 您在控制台创建的Group ID。
        properties.put(PropertyKeyConst.GROUP_ID, config.getGroupId());
        // AccessKey ID阿里云身份验证，在阿里云RAM控制台创建。
        properties.put(PropertyKeyConst.AccessKey, config.getAccessKey());
        // Accesskey Secret阿里云身份验证，在阿里云服RAM控制台创建。
        properties.put(PropertyKeyConst.SecretKey, config.getSecretKey());
        // 设置TCP接入域名，进入控制台的实例详情页面的TCP协议客户端接入点区域查看。
        properties.put(PropertyKeyConst.NAMESRV_ADDR, config.getNameSrvAddr());

        // 集群订阅方式(默认)。
        properties.put(PropertyKeyConst.MessageModel, PropertyValueConst.CLUSTERING);

        //消息生产者配置参数设置
        if (ConfigType.PRODUCER.equals(configType)) {
            Long sendMsgTimeoutMillis = config.getSendMsgTimeoutMillis();
            if (null != sendMsgTimeoutMillis && sendMsgTimeoutMillis > 0) {
                properties.put(PropertyKeyConst.SendMsgTimeoutMillis, sendMsgTimeoutMillis);
            }
        }

        if (ConfigType.CONSUMER.equals(configType)) {
            Integer consumeThreadNums = config.getConsumeThreadNums();
            if (null != consumeThreadNums && consumeThreadNums > 0) {
                properties.put(PropertyKeyConst.ConsumeThreadNums, consumeThreadNums);
            }
            String expression = config.getExpression();
            if (null == expression || expression.trim().length() == 0) {
                throw new IllegalArgumentException("消费者配置中缺少消息过滤表达式");
            }

            Integer maxReconsumeTimes = config.getMaxReconsumeTimes();
            if (null != maxReconsumeTimes && maxReconsumeTimes > 0) {
                properties.put(PropertyKeyConst.MaxReconsumeTimes, maxReconsumeTimes);
            }

            Integer maxCachedMessageAmount = config.getMaxCachedMessageAmount();
            if (null != maxCachedMessageAmount && maxCachedMessageAmount > 0) {
                properties.put(PropertyKeyConst.MaxCachedMessageAmount, maxCachedMessageAmount);
            }
        }

        return properties;
    }

    private enum ConfigType {
        /**
         * 消息生产者
         */
        PRODUCER,
        /**
         * 消息消费者
         */
        CONSUMER
    }
}
